<?php
/**
 * Created by PhpStorm.
 * User: yzh
 * Date: 2019/5/17
 * Time: 10:28
 */

namespace app\common\lib;


use think\Config;
use think\Exception;
use think\Log;

class Queue
{

    private static $_instance;

    public static function getInstance(){
        if (empty(self::$_instance)){
            self::$_instance = new self();
        }
        return self::$_instance;
    }
    /**
     * 生产队列消息
     * @param $destination
     * @param array $msg_data
     * @param bool $persistent
     * @return bool|string
     */
    public function sendToMQ($destination,array $msg_data, $persistent = false)
    {
        try{
            $stomp = new \Stomp('tcp://172.26.207.51:61613');
        }catch (StompException $exception){
            //todo 服务通知，队列服务挂掉，短信，邮件
            Log::write($exception,'error',true);//队列错误
            return 'failure';
        }
        try {
            $stomp->begin("Transaction");
            foreach ($msg_data as $item){
                $stomp->send($destination, json_encode($item), array('persistent'=> $persistent));
            }
            $stomp->commit("Transaction");
            unset($stomp);
            return 'success';
        } catch (\Exception $exception) {
            Log::write($exception,'error',true);//队列错误
            return 'failure';
        }
    }

    /**
     * 消费队列
     * @param $destination
     * @param array $function
     * @param int $tag 次数标识
     * @return string
     */
    public function getMessageFormQueue($destination,array $function,$tag=1)
    {
        try {
            $stomp = new \Stomp('tcp://172.26.207.51:61613');
        } catch (StompException $exception) {
            //todo 服务通知，队列服务挂掉，短信，邮件
            Log::write($exception, 'error', true);//队列错误
            return 'failure';
        }
        $stomp->subscribe($destination);//订阅队列
        try {
            while ($stomp->hasFrame()) {
                //查看是否还有消息
                if ($stomp->hasFrame()) {
                    $frame = $stomp->readFrame();
                    try {
                        call_user_func_array($function, ['data' => json_decode($frame->body, true)]);//调用外部任务函数
                    } catch (\Exception $exception) {
                        Log::write($exception, 'error', true);//队列错误
                        //同样也要通知队列已经处理过该消息
                        $stomp->ack($frame);
                        continue;//跳出本次循环
                    }
                    //最后通知mq，我们已经处理了该条消息
                    $stomp->ack($frame);
                }
            }
            $stomp->unsubscribe($destination);//移除队列订阅
            unset($stomp);//断开连接
        }catch (\Exception $exception){//队列报错，重新请求消费队列
            Log::write($exception->getMessage(), 'error', true);//队列错误
            if ($tag<=3){//发送3次推送尝试
                $this->getMessageFormQueue($destination,$function,$tag+1);//重新发起队列推送
            }
            return 'failure';
        }
    }
}